
/****************************************************************************
**
** Copyright (C) 2004-2008 VATATA.com . All rights reserved.
**
** This file is the part of the Miniutil Poject of VATATA.com.
** Vatata's miniutil library is lightweight and portable (for linux
** and Windows). The C++ Class Library included Threadpool, Dll danymic loading, 
** General socket application model and tcp server, httpserver/servlet, etc. 
** It is a pratical utility disigned for who don't want learn the Big C++ 
** Developing Platform, such as  Boost or ACE.
**
** This file may be used under the terms of the GNU General Public
** License versions 2.0 or 3.0 as published by the Free Software
** Foundation and appearing in the files LICENSE.GPL2 and LICENSE.GPL3
** included in the packaging of this file.  Alternatively you may (at
** your option) use any later version of the GNU General Public
** License if such license has been publicly approved by VATATA.com.
** (or its successors, if any) and the Miniutil Project Aministrators. 
**
** Please review the following information to ensure GNU General
** Public Licensing requirements will be met:
** http://www.gnu.org/licenses/lgpl.html. If you are unsure which 
** license is appropriate for your use, please review the following 
** information: http://code.google.com/p/miniutil/ or contact 
** http://www.vatata.com.
**
** In addition, as a special exception, VATATA.com, as the sole
** copyright holder for Miniutil Project, grants users of VATATA
** P2P Platform the right to release and package with the related 
** libraries without the source code.
**
** This file is provided "AS IS" with NO WARRANTY OF ANY KIND,
** INCLUDING THE WARRANTIES OF DESIGN, MERCHANTABILITY AND FITNESS FOR
** A PARTICULAR PURPOSE. vatata reserves all rights not expressly
** granted herein.
**
** This file is provided AS IS with NO WARRANTY OF ANY KIND, INCLUDING THE
** WARRANTY OF DESIGN, MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE.
**
****************************************************************************/


#include <ctime>
#include "miniutil/asyncnet.h"
#include "miniutil/threadpool.h"
#include "miniutil/date.h"
#include "miniutil/log.h"

#include "miniutil/strutil.h"

#ifndef WIN32
#include <errno.h>
#endif

using namespace miniutil::asyncnet;

int socket_buffer::put_to_buffer( char * data, int len )
{
    if ( len == -1 )
        len = (int) strlen( data );
    else if ( len == 0 )
        return 0;

    miniutil::auto_lock lk( buf_mutex );

    w_buf.enlarge( len );

    w_buf.append( data, (size_t) len );

    return len;
}


session::session()
{
	session_stus = ASYN_STUS_IDLE;
}

session::~session()
{
    close();
}

void session::close()
{
	session_stus = ASYN_STUS_FINI;
    //async_selector::get_selector()->remove( this );
}

int session::on_connected()
{
    //AUTOENTRY
	session_stus = ASYN_STUS_CONNED;
	return 0;
}

int session::on_accepted()
{
    on_connected();
	session_stus = ASYN_STUS_CONNED;
	return 0;
}

int session::on_disconnected()
{
 	session_stus = ASYN_STUS_FAIL;
   //AUTOENTRY
	return 0;
}

int session::on_exception( const miniutil::socketException & e )
{
	session_stus = ASYN_STUS_FAIL;
    L_DEBUG("session::on_exception( const miniutil::socketException : %s\n", e.msg.c_str() );
	return 0;
}

int session::on_recv( )
{
    //AUTOENTRY
	return 0;
}

int session::on_sendover( int len )
{
    AUTOENTRY
	return 0;
}


//-------------------monitor----------------
/*
on_connected(socket)
on_accepted(socket)

return: 
 1  socket is used by other object
 2  socket is of no use, can close it
 
on_timeout(ip, port)
on_fail(ip, port)
	0 
	1 

090407: NOT IMPLEMENT FULLY!!! be care
monitor not implent yet, should delete this before return 1
libdown3's btpeer_conn_monitor do like above spec.
*/

monitor::monitor()
{
	moni_stus = ASYN_STUS_IDLE;
}

monitor::~monitor()
{
}

//bool monitor::connect(const char *host, int port, int timeout_ms )
//{
//	miniutil::ip4addr hostip(host);
//    miniutil::soaddr peer( port,hostip );
//
//	SOCKET thesock = ::socket(PF_INET, SOCK_STREAM, 0);
//
//	if( thesock <= 0 ) 
//        return false;
//
//    #if defined(WIN32)
//	    u_long blockmode = 1;
//	    int ret = ioctlsocket(thesock, FIONBIO, &blockmode);
//        if ( ret == SOCKET_ERROR) return NULL;
//    #else
//	    fcntl(thesock, F_SETFL,  O_NONBLOCK );
//    #endif
//
//    int res = ::connect( thesock, peer.get_sockaddr(), sizeof(struct sockaddr) );
//	if(res < 0)
//	{
//#ifdef WIN32
//		if(miniutil::socket::getLastError() != WSAEWOULDBLOCK)
//#else
//		if ( miniutil::socket::getLastError() != EINPROGRESS )
//#endif
//			return false;
//	}
//
//    //miniutil::socket * p_sock = new miniutil::socket( thesock );
//
//    return async_selector::get_selector()->put(host, port, thesock, this , timeout );
//}

bool monitor::connect(const char *host, int port, int timeout_ms )
{
	int ret = async_selector::get_selector()->newconnectinfo(host, port, this, timeout_ms);
	return true;
}


int monitor::on_connected( miniutil::socket * p_sock ) 
{
    L_DEBUG( "monitor::on_connected %s %d\n", p_sock->peerIP().IP().c_str(), p_sock->peerPort() );
	moni_stus = ASYN_STUS_CONNED;
	return 1;
}

int  monitor::on_connected_fail( const char * host, int port )
{
    L_DEBUG( "monitor::on_connected_fail %s %d\n", host, port );
	moni_stus = ASYN_STUS_FAIL;
	return 1;
}

int monitor::on_connected_timeout( const char * host, int port )
{
    L_DEBUG( "monitor::on_connected_timeout %s %d\n", host, port );
	moni_stus = ASYN_STUS_FAIL;
	return 1;
}

bool monitor::bind( int port, const miniutil::ip4addr & ip )
{
    try{
        miniutil::serversocket * p_ssocket = new miniutil::serversocket( port, ip );

        return async_selector::get_selector()->put( p_ssocket, this );
    }catch(...)
    {}
    return false;
}

void monitor::remove( int port , const miniutil::ip4addr & ip  )
{
    async_selector::get_selector()->remove( port, ip );
}

void monitor::remove( const char * host, int port  )
{
    async_selector::get_selector()->remove( host , port );
}


int monitor::on_accepted( miniutil::socket * p_sock )
{
    L_DEBUG( "monitor::on_accept %s %d\n", p_sock->peerIP().IP().c_str(), p_sock->peerPort() );
	moni_stus = ASYN_STUS_CONNED;
	return 0;
}


namespace miniutil
{
    namespace asyncnet
    {
        struct connector_info
        {
            string                              host;
            int                                 port;
            SOCKET                              sock;
            monitor *                           p_con;
            miniutil::m_uint32_t                timeout_ms;
            miniutil::m_uint32_t                connect_time;

            inline
            connector_info(const char * _host, int _port, SOCKET s,  monitor * p , int  tmout_ms) 
                : host( _host )
                , port ( _port )
                , p_con( p )
                , sock( s )
                , connect_time( miniutil::date::uptime_ms()  ) 
            {
                timeout_ms = tmout_ms > 0 ? (miniutil::m_uint32_t) tmout_ms : 1000 * 120; 
            }

			~connector_info()
			{
			}
        };

        struct acceptor_info
        {
            miniutil::serversocket *        p_ssock;
            monitor *                       p_acceptor;
            SOCKET                          ssock;

            inline
            acceptor_info( miniutil::serversocket * p_ss, monitor * p )
                : p_ssock( p_ss )
                , p_acceptor( p )
                , ssock( p_ss->get_sock() )
            {}
        };

        struct session_info
        {
            miniutil::socket *          p_sock;
            session *                   p_session;
            SOCKET                      sock;
            socket_buffer *             p_buffers;

            inline
            session_info( miniutil::socket * psock, session * psession, socket_buffer * pbuf ) 
            : p_sock(psock )
            , p_session( psession )
            , sock( p_sock->get_sock() )
            , p_buffers( pbuf )
            {
            }
        };

        class async_selector_select : public async_selector , public miniutil::runnable
        {
            miniutil::mutex                 selector_mutex;
            vector<connector_info * >       connector_array;
            vector<acceptor_info * >        acceptor_array;
            vector<session_info * >         session_array;
        public:
            int timeout_ms;
            bool is_running;

        public:
            async_selector_select();
            void run();

            int init_fdset( fd_set * p_rset, fd_set * p_wset, fd_set * p_eset);

            //void do_connected(fd_set * p_wset );
			void do_connecter(fd_set * p_wset, fd_set * p_eset);
            void do_accepted( fd_set * p_rset );

            void do_send(fd_set * p_wset );
            void do_recv( fd_set * p_rset );

			void do_exception(fd_set * p_wset);

            virtual bool put( const char * host, int port, SOCKET sock  , monitor * con , int timeout_ms ) ;
            virtual void remove(  const char * host, int port ) ;

            virtual bool put( miniutil::serversocket * p_ssock, monitor * p_srv ) ;
            virtual void remove( int port , const miniutil::ip4addr & ip  ) ;

            virtual bool put( session * p_session , miniutil::socket * p_sock) ;
            virtual void remove( session * p_session ) ;

            void remove_session( int k );
		public:	
			struct connectinfo
			{
				std::string host;
				int port;
				monitor* mon;
				int timeout_ms;
			};
			virtual int newconnectinfo(const char* host, int port, monitor *con, int timeout_ms);
			int checknewconninfo();
			miniutil::mutex conninfo_mutex;
			std::vector<connectinfo> vconninfos;

        };
    }
}

async_selector * async_selector::get_selector()
{
    static async_selector * p = NULL ;
    if ( p == NULL )
        p = new async_selector_select();
    return p;
}

async_selector_select::async_selector_select()
: timeout_ms( 5000 )
, is_running( false )
{
}

void async_selector_select::run()
{
	struct timeval timeout;
	timeout.tv_sec = timeout_ms / 1000;
	timeout.tv_usec = (timeout_ms % 1000) * 1000 ;
    fd_set recv_set, send_set, exception_set;

	this->is_running = true;
	time_t now, pre = 0;
    while( this->is_running  )
    {
		time(&now);
		if(now > pre)
		{
		    M_DEBUG( "async_selector_select::run() working...\n");
			pre = now;
		}
		checknewconninfo();

		selector_mutex.lock();
        int max_socket = this->init_fdset( & recv_set, & send_set, &exception_set);
		selector_mutex.unlock();

        if ( max_socket == 0 )
        {
		    M_DEBUG( "async_selector_select thread finished by no socket!, and it will restart when it is put another socket!\n");
            break;
        }
		
        int res = ::select( max_socket + 1, &recv_set, &send_set, &exception_set, &timeout ); 
	    if ( res < 0 )
	    {
#ifdef WIN32
			int sret = WSAGetLastError();
#endif
		    thread::sleep( 10 );
		    continue;
	    }
	    else if ( res == 0 ) 
		{
		    thread::sleep( 10 );
		    continue;
		}
		selector_mutex.lock();
	    this->do_recv( &recv_set );
	    this->do_send( &send_set );
        this->do_accepted( &recv_set );
		this->do_connecter( &send_set, &exception_set);
	    //this->do_connected( &send_set );
		//this->do_exception( &exception_set);
        selector_mutex.unlock();
    }
	selector_mutex.lock();
    is_running = false;
    selector_mutex.unlock();
}

void async_selector_select::remove_session( int k )
{
	AUTOENTRY
    session_info * pinfo = session_array[k];
    L_DEBUG("async_selector_select::remove_session %d %p\n", k, pinfo);

    session_array[k] = NULL;

    pinfo->p_sock->close();
    //pinfo->p_session->on_disconnected();
    delete pinfo->p_sock;
    delete pinfo;
}

void async_selector_select::do_recv(fd_set * p_rset )
{
    static int count = 0;
    int size = (int) session_array.size();
    if ( size == 0 ) return;

    if ( count++ > size ) count = count % size;

    for( int i = 0; i<size; i++ )
    {
        int k =( i + count ) % size;
        session_info * pinfo = session_array[k];
        if ( pinfo == NULL ) continue;

        if ( ! FD_ISSET( pinfo->sock, p_rset) )
            continue;

        miniutil::auto_lock lk( pinfo->p_buffers->buf_mutex );

        miniutil::data::buffer & buf = pinfo->p_buffers->r_buf;
        if ( buf.free_length() < 1024 )
            buf.resize( buf.size() );

        int recv_len = ::recv( pinfo->sock, buf.tail(), (int) buf.free_length(), 0 );
        if ( recv_len > 0 )
        {
            buf.inc_length( recv_len );
            int ret = pinfo->p_session->on_recv();
			if(ret == 1)
				this->remove_session( k );
        }
        else if ( recv_len < 0 && miniutil::socket::getLastError() == EWOULDBLOCK )
        {
            /// impossible ?
        }
        else 
        {
            if ( recv_len < 0 )
            {
                pinfo->p_session->on_exception( miniutil::socketException() );
            }
			else if(recv_len == 0)
			{
				pinfo->p_session->on_disconnected();
			}
            this->remove_session( k );
        }

    }
}

void async_selector_select::do_send(fd_set * p_wset )
{
    static int count = 0;
    int size = (int) session_array.size();
    if ( size == 0 ) return;

    if ( count++ > size ) count = count % size;

    for( int i = 0; i<size; i++ )
    {
        int k =( i + count ) % size;
        session_info * pinfo = session_array[k];
        if ( pinfo == NULL ) continue;

        if ( ! FD_ISSET( pinfo->sock, p_wset) )
            continue;

        miniutil::auto_lock lk( pinfo->p_buffers->buf_mutex );
        miniutil::data::buffer & buf = pinfo->p_buffers->w_buf;

        int send_len = ::send( pinfo->sock, buf.data(), (int)buf.length(), 0 );
        if ( send_len > 0 )
        {
            if ( send_len == (int) buf.length() )
			{
                int ret = pinfo->p_session->on_sendover( send_len );
				if(ret == 1)
				{
					 this->remove_session( k );
				}
			}
            buf.shift( send_len );
        }
        else if ( send_len < 0 && miniutil::socket::getLastError() == EWOULDBLOCK )
        {
		}
		else
		{
            if ( send_len < 0 )
            {
                pinfo->p_session->on_exception( miniutil::socketException() );
            }
			else if(send_len == 0)
			{
				pinfo->p_session->on_disconnected();
			}
            this->remove_session( k );

            //pinfo->p_session->on_exception( NULL );
            //L_DEBUG("miniutil::socketException::trygeterrormessage() %s\r\n", miniutil::socketException::trygeterrormessage().c_str() );
            //this->remove_session( k );
        }
    }
}


void async_selector_select::do_accepted(fd_set * p_rset )
{
	try{
    for( int i= (int)(acceptor_array.size() -1 ); i>=0 ; i-- )
    {
        if ( acceptor_array[i] == NULL )
            continue;
        acceptor_info * p_ainfo = acceptor_array[i] ;
        if (  FD_ISSET(p_ainfo->ssock, p_rset ) )
        {
            SOCKET sock = ::accept( p_ainfo->ssock, NULL , NULL );
	        if( sock > 0 )
            {
                int ret = p_ainfo->p_acceptor->on_accepted( new miniutil::socket( sock ) );
				if(ret == 2)
				{
					closesocket(sock);
				}
				acceptor_array[i] = NULL ;
				delete p_ainfo;
            }
        }
    }
	}
	catch(...)
	{
		M_DEBUG("do_accepted exception...");
	}
}

//void async_selector_select::do_connected(fd_set * p_wset )
void async_selector_select::do_connecter(fd_set * p_wset, fd_set * p_eset)
{
	try
	{
    for( int i = (int)( connector_array.size() - 1); i >= 0 ; i-- )
    {
        if ( connector_array[i] == NULL )
            continue;

        connector_info *pinfo = connector_array[i];
        if ( FD_ISSET( pinfo->sock, p_wset ) )
        {
			miniutil::socket * p_sock =NULL;
			try{
             p_sock = new miniutil::socket( pinfo->sock );
			}
			catch(socketException ex)
			{
				p_sock = NULL;
				M_DEBUG("%s\n", ex.msg.c_str());
			}
			catch(...)
			{
				p_sock = NULL;
			}
			//if(p_sock != NULL)
			{
				int ret = pinfo->p_con->on_connected( p_sock );
				if(ret == 2)
				{
					closesocket(pinfo->sock);
				}
				connector_array[i] = NULL ;
				delete pinfo;
			}
			continue;
        }

        if ( FD_ISSET( pinfo->sock, p_eset ) )
        {
			int ret = pinfo->p_con->on_connected_fail( pinfo->host.c_str(), pinfo->port );
			closesocket(pinfo->sock);
            connector_array[i] = NULL ;
            delete pinfo;
			continue;
        }

    }
	}
	catch(...)
	{
		M_DEBUG("do_connected exception...");
	}
}

void async_selector_select::do_exception(fd_set * p_eset)
{
    for( int i = (int)( connector_array.size() - 1); i >= 0 ; i-- )
    {
        if ( connector_array[i] == NULL )
            continue;

        connector_info *pinfo = connector_array[i];
        if ( FD_ISSET( pinfo->sock, p_eset ) )
        {
			int ret = pinfo->p_con->on_connected_fail( pinfo->host.c_str(), pinfo->port );
			closesocket(pinfo->sock);
            connector_array[i] = NULL ;
            delete pinfo;
        }
    }
}

int async_selector_select::init_fdset( fd_set * p_rset, fd_set * p_wset, fd_set * p_eset )
{
    FD_ZERO( p_rset );
    FD_ZERO( p_wset );
	FD_ZERO( p_eset );
    int max_socket = 0;
    
    for( int i = (int)( connector_array.size() - 1); i >= 0 ; i-- )
    {
        connector_info *pinfo = connector_array[i];
        if ( pinfo == NULL )  continue;

		if(pinfo->p_con == NULL)
		{
            connector_array[i] = NULL ;
            delete pinfo;
            continue;
		}

        //L_DEBUG("check timeout %d %d %d\n", pinfo->connect_time , pinfo->timeout_ms ,miniutil::date::uptime_ms() );
        if ( pinfo->connect_time + pinfo->timeout_ms < miniutil::date::uptime_ms() )
        {
            pinfo->p_con->on_connected_timeout( pinfo->host.c_str(), pinfo->port );
			closesocket(pinfo->sock);
            connector_array[i] = NULL ;
            delete pinfo;
            continue;
        }

        max_socket = max( max_socket,(int) pinfo->sock );
        FD_SET( pinfo->sock, p_wset );
		FD_SET( pinfo->sock, p_eset );
    }

    for( int j= (int)(acceptor_array.size() -1 ); j>=0 ; j-- )
    {
        if ( acceptor_array[j] == NULL )
            continue;
        FD_SET( acceptor_array[j]->ssock, p_rset );
    }

    for( int k= (int)(session_array.size() - 1 ); k>=0 ; k-- )
    {
        session_info * psinfo = session_array[k];
        if ( psinfo == NULL ) continue;

        FD_SET( psinfo->sock, p_rset );

        max_socket = max( max_socket,(int) (psinfo->sock) );

        if ( psinfo->p_buffers->w_buf.length() > 0 )
            FD_SET( psinfo->sock, p_wset );
    }

    return max_socket;
}

bool async_selector_select::put(const char * _host, int _port, SOCKET sock , miniutil::asyncnet::monitor *con, int timeout_ms )
{
   if (  con == NULL || sock <= 0 )
        return false;

    miniutil::auto_lock selector_lock( this->selector_mutex );

    connector_info * p_info = new connector_info( _host, _port, sock, con,  timeout_ms );

    for( int i = (int)( connector_array.size() - 1); i >= 0 ; i-- )
    {
        if ( connector_array[i] == NULL )
        {
            connector_array[i] = p_info;
            p_info = NULL;
            break;
        }
    }
    if ( p_info ) connector_array.push_back( p_info );

    if ( this->is_running == false )
    {
		M_DEBUG("async_selector_select thread start\n");
		this->is_running = miniutil::threadpool::RunTask( this );
    }

    return true;
}

void async_selector_select::remove( const char * host, int port ) 
{
    miniutil::auto_lock selector_lock( this->selector_mutex );

    for( int i = (int)( connector_array.size() - 1); i >= 0 ; i-- )
    {
        connector_info * pinfo = connector_array[i];
        if ( pinfo == NULL )
            continue;

        if ( pinfo->port == port && strnicmp( host, pinfo->host.c_str(), pinfo->host.length() ) == 0 )
        {
            connector_array[i] = NULL;
            delete pinfo;
        }
    }
} 

bool async_selector_select::put( miniutil::serversocket * p_ssock, monitor * p_acceptor ) 
{
    if ( p_ssock == NULL || p_acceptor == NULL )
        return false;

    miniutil::auto_lock selector_lock( this->selector_mutex );

    acceptor_info * p_ainfo = new acceptor_info( p_ssock, p_acceptor );
    for( int i= (int)(acceptor_array.size() -1 ); i>=0 ; i-- )
    {
        if ( acceptor_array[i] == NULL )
        {
            acceptor_array[i] = p_ainfo;
            p_ainfo = NULL;
            break;
        }
    }
    if ( p_ainfo ) acceptor_array.push_back( p_ainfo );

    if ( this->is_running == false )
    {
        this->is_running = miniutil::threadpool::RunTask( this );
    }

    return true;
}

void async_selector_select::remove( int port , const miniutil::ip4addr & ip  ) 
{
    miniutil::auto_lock selector_lock( this->selector_mutex );

    for( int i= (int)(acceptor_array.size() -1 ); i>=0 ; i-- )
    {
        if ( acceptor_array[i] == NULL )
            continue;

        serversocket * p_ssock = acceptor_array[i]->p_ssock;
        if ( p_ssock == NULL || ( p_ssock->getPort() == port && p_ssock->getIP() == ip ) )
        {
            delete acceptor_array[i];
            acceptor_array[i] = NULL;

            p_ssock->close();
            continue;
        }
    }
}

bool async_selector_select::put( session * p_session , miniutil::socket * p_sock) 
{
    if ( p_session == NULL || p_sock == NULL || p_sock->isClosed() )
        return false;

    miniutil::auto_lock selector_lock( this->selector_mutex );
    session_info * p_sinfo = new session_info( p_sock, p_session , &( p_session->sock_buf ) );

    for( int i= (int)(session_array.size() -1 ); i>=0 ; i-- )
    {
        if ( session_array[i] == NULL )
        {
            session_array[i] = p_sinfo;
            p_sinfo = NULL;
            break;
        }
    }
    if ( p_sinfo ) session_array.push_back( p_sinfo );

    if ( this->is_running == false )
    {
        this->is_running = miniutil::threadpool::RunTask( this );
    }

    return true;
}

void async_selector_select::remove( session * p_session ) 
{
    if ( p_session == NULL ) return;

    miniutil::auto_lock selector_lock( this->selector_mutex );

    for( int i= (int)(session_array.size() -1 ); i>=0 ; i-- )
    {
        if ( session_array[i] == NULL )
            continue;
        if ( session_array[i]->p_session == p_session )
        {
            session_info * pinfo = session_array[i];
            session_array[i] = NULL;

            pinfo->p_sock->close();
            delete pinfo->p_sock;
            delete pinfo;
        }
    }
}

int async_selector_select::newconnectinfo(const char* host, int port, monitor *con, int timeout_ms)
{
	connectinfo info;
	info.host = host;
	info.port = port;
	info.mon = con;
	info.timeout_ms = timeout_ms;

	conninfo_mutex.lock();
	vconninfos.push_back(info);
	conninfo_mutex.unlock();

	miniutil::auto_lock selector_lock( this->selector_mutex );
	if ( this->is_running == false )
	{
		this->is_running = miniutil::threadpool::RunTask( this );
	}
	return 0;
}

int async_selector_select::checknewconninfo()
{
	miniutil::auto_lock lk(&conninfo_mutex);
	if(vconninfos.size() == 0)
		return 0;

	for(unsigned int i=0; i< vconninfos.size(); i++)
	{
		connectinfo info = vconninfos[i];

		miniutil::ip4addr hostip(info.host);
		miniutil::soaddr peer( info.port,hostip );

		SOCKET thesock = ::socket(PF_INET, SOCK_STREAM, 0);

		if( thesock <= 0 ) 
			return false;

		#if defined(WIN32)
			u_long blockmode = 1;
			int ret = ioctlsocket(thesock, FIONBIO, &blockmode);
			if ( ret == SOCKET_ERROR) return NULL;
		#else
			fcntl(thesock, F_SETFL,  O_NONBLOCK );
		#endif

		int res = ::connect( thesock, peer.get_sockaddr(), sizeof(struct sockaddr) );
		if(res < 0)
		{
	#ifdef WIN32
			if(miniutil::socket::getLastError() != WSAEWOULDBLOCK)
	#else
			if ( miniutil::socket::getLastError() != EINPROGRESS )
	#endif
			{
				//return false;
				M_DEBUG("::connect fail..\n");
				info.mon->on_connected_fail(info.host.c_str(), info.port);
				closesocket(thesock);
				continue;
			}
		}
		this->put(info.host.c_str(), info.port, thesock, info.mon , info.timeout_ms );
	}
	vconninfos.clear();
	return 0;

}

